package org.fujene.index;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import org.fujene.communicate.repli.ReplicatePeerSystem;
import org.fujene.structure.Constants;
import org.fujene.structure.ExtendQueue;
import org.msgpack.MessagePackObject;
import org.msgpack.object.NilType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Request;
import org.msgpack.rpc.Server;
import org.msgpack.rpc.loop.EventLoop;

/**
 * インデックス時の窓口として機能します。起動時に指定したデータ構造を入力することができます。
 * 
 * @author fujisaka_yusuke
 *
 */
public class ReplicateIndexSystem {
	public boolean stopFlag;
	private ReplicatePeerSystem peer;
	private ExecutorService exec = null;
	private int contentTypes, supplementTypes;
	private Logger logger;
	
	public ReplicateIndexSystem(ReplicatePeerSystem peer, int content, int supplement, Logger logger) {
		exec = Executors.newSingleThreadExecutor();
		
		this.peer = peer;
		this.contentTypes = content;
		this.supplementTypes = supplement;
		this.logger = logger;
	}
	
	// RPC対応
	public class IndexRPCHandler {
		public void index(Request req) throws Exception {
			MessagePackObject obj = req.getArguments();
			MessagePackObject[] args = obj.asArray();
			long ID;
			String[] content;
			String[] supplement;
			
			int requireLength = contentTypes + supplementTypes + 1;
			if (args.length < requireLength || !args[0].isIntegerType()) {
				req.sendError(RawType.create("Error(index): Not efficient index word or ID"));
				logger.severe("Error(index): Not efficient index word or ID");
			}

			// つけるIDがなければ自前で付けておく
			ID = args[0].asLong();
			
			content = new String[contentTypes];
			supplement = new String[supplementTypes];
			for (int i = 0; i < content.length; i++) {
				content[i] = args[i+1].asString();
			}
			for (int i = 0; i < supplement.length; i++) {
				supplement[i] = args[i+1+content.length].asString();
			}
			
			peer.setNewIndexExtent(new ExtendQueue(content, supplement, ID));

			req.sendResult(NilType.create());
		}

		public void read(Request req) {
			peer.readMemory();
			req.sendResult(NilType.create());
		}

		public void save(Request req) {
			peer.saveMemory();
			req.sendResult(NilType.create());
		}

		public void stop(Request req) {
			peer.setStopper(true);
			this.notifyAll();
			exec.shutdownNow();
			req.sendResult(NilType.create());
		}
	}

	public class IndexMainService implements Runnable {
		public IndexMainService() {
		}

		@Override
		public void run() {
			logger.fine("IndexSystem started.");
			// RPC対応
			Server server = new Server(EventLoop.defaultEventLoop());
			try {
				server.serve(new IndexRPCHandler());
				server.listen(Constants.INDEX_PORT);

				logger.fine("Index server port " + Constants.INDEX_PORT
						+ " has established.");

				synchronized (this) {
					this.wait();
				}

				logger.fine("IndexMainService ended.");

				server.close();
			} catch (InterruptedException ignored) {
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
}
